package rm3;


import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.PriorityQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import server.ServerImpl;

public class UDPManagerListener extends Thread
{
	DatagramSocket listenerSocket = null;
	byte[] requested, ack;
	public static ServerImpl montreal;
	public static ServerImpl vancouver;
	public static ServerImpl toronto;
	ExecutorService serverthreadPool;
	long [] arrayDuplicated;
	int ArrayIndex = 0;
	long sequencer;
	public static ConcurrentHashMap<Long, String> storedReply = new ConcurrentHashMap <Long, String> ();

     

	public UDPManagerListener(DatagramSocket socket, ServerImpl montrealObj, ServerImpl torontoObj, ServerImpl vancouverObj){
		listenerSocket = socket;
		montreal = montrealObj;
		toronto = torontoObj;
		vancouver = vancouverObj;
		serverthreadPool = Executors.newFixedThreadPool(40); 
		arrayDuplicated = new long[1000] ;
		sequencer = 1;
	}

	public void run() {
		try{
			System.out.println("Replica 3 is listening to the replica manager");
			while(true){
				requested = new byte[100];
				ack = new byte[16];
				// receive client's request
				DatagramPacket clientRequest = new DatagramPacket(requested, requested.length);
				listenerSocket.receive(clientRequest); 
				
				// send to holdback queue
				try {
					replyAndReceiveAck(clientRequest);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (ExecutionException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				
				MRMessage m = new MRMessage();
				m.parseMessage(new String (clientRequest.getData(),0, clientRequest.getLength()));
				System.out.println("service requested at RM3: "+  (new String (clientRequest.getData(),0, clientRequest.getLength())));
				
				if(m.getRequestType().equals("reply"))
					sendReply(m.getSequencer(), clientRequest, listenerSocket);
				
				else {
				ack =  Integer.toString(m.getSequencer()).getBytes();
				
				DatagramPacket replytoClient = new DatagramPacket(ack, ack.length,
						clientRequest.getAddress(), clientRequest.getPort());

				listenerSocket.send(replytoClient);
				}
			}

		}catch(SocketException e){
			System.out.println("Server Socket Exception: " + e.getMessage());
		}catch (IOException e) {
			System.out.println("IO: " + e.getMessage());
		}finally {
			if(listenerSocket != null) listenerSocket.close();
		}
		System.out.println("Exiting server's try catch!");

	}
	
	public void sendReply(int id, DatagramPacket client, DatagramSocket server){
		
		byte[] replyFromreplica = new byte[16];
		DatagramPacket tosend;
		if(sequencer>id){
			System.out.println("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$");
			replyFromreplica = storedReply.get(id).getBytes();
			tosend = new DatagramPacket(replyFromreplica, replyFromreplica.length,
					client.getAddress(), client.getPort());
			
		} else {
			System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
			replyFromreplica ="999999".getBytes();
			tosend = new DatagramPacket(replyFromreplica, replyFromreplica.length,
					client.getAddress(), client.getPort());
		}
		try {
			server.send(tosend);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	
	
	
	public void replyAndReceiveAck(DatagramPacket clientRequest) throws InterruptedException, ExecutionException, SocketException, UnknownHostException{
		byte[] replyToreplicaM = new byte[100];
		DatagramPacket tosend;
		DatagramPacket client;
		DatagramSocket serverSocket = new   DatagramSocket();
		MRMessage m = new MRMessage();
		m.parseMessage((new String (clientRequest.getData(),0, clientRequest.getLength())));
		boolean flag = false;
		String answer = "-99";
		MRMessage replymesssage ;
		String replyToReplicaManager = " ";
		//if(!IsDuplicated(m))
		//{
			if(m.getRequestType().compareTo("update") == 0)
			{
				//arrayDuplicated[ArrayIndex] = m.getSequencer();
				//ArrayIndex++;
				//sequencer++;
				UpDataServer(m);
				replyToReplicaManager = "done";
			}
			else if(m.getRequestType().compareTo("getstate") == 0)
			{
				replymesssage = getStateServer();
				replymesssage.setRequestType("update");
				replyToReplicaManager = replymesssage.getPacket();
				//arrayDuplicated[ArrayIndex] = m.getSequencer();
				//ArrayIndex++;
				//sequencer++;
			}
				
		//}
		//else 
		//{
			//System.out.println(">>>>>> Request is Duplicated , just ignor" + m.getSequencer());
		
		//}
			
			
			replyToreplicaM = replyToReplicaManager.getBytes();
				tosend = new DatagramPacket(replyToreplicaM, replyToreplicaM.length,
						clientRequest.getAddress(), clientRequest.getPort());

			
			try {
				serverSocket.send(tosend);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

	}
	

	
	/**
	 * this method is used to up data the servers
	 * @param message
	 * @return
	 */
	public String UpDataServer(MRMessage message){
		String result = "";
		montreal.inventory.upDate(message.getMontreal());
		vancouver.inventory.upDate(message.getVancouver());
		toronto.inventory.upDate(message.getToronto());
		
		System.out.println("Replica has been up data it "+ result);
		return result;
	}
	
	/**
	 * this method is used to get the state of the servers
	 * @return
	 */
	public MRMessage getStateServer()
	{
		MRMessage message = new MRMessage();
		
		message.setMontreal(montreal.inventory.getState());
		message.setVancouver(vancouver.inventory.getState());
		message.setToronto(toronto.inventory.getState());
		return message;
	}
	
	
	public boolean IsDuplicated(MRMessage message)
	{
		MRMessage localmessage = new MRMessage();
		localmessage = message;
		boolean Duplicated = false;
		for(int i = 0; i < arrayDuplicated.length; i++) 
		{  
			
			if(arrayDuplicated[i] == localmessage.getSequencer())
			{
				Duplicated = true;
			}
		}  
		return Duplicated;
	}

}

